/*
 * @Descripttion: 
 * @version: 
 * @Author: jiuguo.qin
 * @Date: 2019-12-22 18:53:25
 * @LastEditors: jiuguo.qin
 * @LastEditTime: 2020-05-18 14:30:23
 */

#include <string>
#include <iostream>
#include <map>
#include <vector>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>

#include "units.h"
#include <sys/time.h>
#include "wrk_record.h"
#include "wrk_logger.h"
#include "server.h"
#include "include/zmalloc.h"
#include <atomic>
#include "wrk_core.h"
// #include "include/stats.h"
#include <math.h>


using namespace std;


extern uint64_t time_us();

WrkRecord::WrkRecord() {
    memset(&record_total, 0, sizeof(wrk_record_t));
    statistics.latency = stats_alloc(WrkCore::cfg.timeout * 1000); //us, WrkCore::cfg.timeout(ms)
    // statistics.requests = stats_alloc(MAX_THREAD_RATE_S);
}

WrkRecord::~WrkRecord() {
    CDEBUG << "rss = "<< format_binary(zmalloc_get_rss()) << std::endl;
    if(WrkCore::cfg.timeout>0) stats_free(statistics.latency);
    // stats_free(statistics.requests);
    
    // delete record_s_ptr;
    cout.flags(ios::fixed);
    cout.precision(4);
    map<string, series >::iterator it_series;
    for(it_series=map_detail_data.begin(); it_series != map_detail_data.end(); it_series++) {
        cout<<"path:"<<it_series->first<<endl;

        series nodes = it_series->second;
        series::iterator it_nodes;
        for(it_nodes = nodes.begin(); it_nodes!=nodes.end(); it_nodes++) {
            stats_data *node = it_nodes->second;
            delete node;
        }
    }

    CDEBUG << "rss = "<< format_binary(zmalloc_get_rss()) << std::endl;
    // 每个path的临时计算结点释放
    map<string, struct interval_point *>::iterator iter;
    for(iter=interval_data.begin(); iter!=interval_data.end(); iter++) {
        if(map_sta.count(iter->first)>0 && map_sta[iter->first]->requests) zfree(map_sta[iter->first]->requests);
        if(map_sta.count(iter->first)>0 && map_sta[iter->first]->latency) zfree(map_sta[iter->first]->latency);

        delete iter->second;
    }

    
    CDEBUG << "rss = "<< format_binary(zmalloc_get_rss()) << std::endl;
}


uint64_t WrkRecord::time_us() {
    struct timeval t;
    gettimeofday(&t, NULL);
    return (t.tv_sec * 1000000) + t.tv_usec;
}


string WrkRecord::get_moment_detail() {
    try {
        stringstream ss;
        ss << "{";
        map<string, wrk_record_t>::iterator iter;
        for(iter=map_tmp_data.begin(); iter!=map_tmp_data.end(); iter++) {
            wrk_record_t tmp_node = iter->second; //统计时间段内的累计值
            string path = iter->first;
            if(path=="" || tmp_node.requests<=0) continue;
                
            if(map_sta.count(path)==0) {
                map_sta[path] = new statistics_t();
                // map_sta[path]->requests = stats_alloc(MAX_THREAD_RATE_S);
                map_sta[path]->latency = stats_alloc(WrkCore::cfg.timeout * 1000);
            }
            
            uint64_t elapsed_ms = time_us() - tmp_node.ctime;
            uint64_t qps = (tmp_node.requests / (double) elapsed_ms) * 1000000;
            stats_record(map_sta[path]->latency, tmp_node.rt/tmp_node.requests);
            
            ss << "\"" << path << "\":{";
            // 连接数用全局 record_total.connections
            ss << "\"connections\":"<< record_total.connections << ",";
            ss << "\"qps\":"<< qps << ",";
            ss << "\"latency\":"<< tmp_node.rt/tmp_node.requests << ",";
            ss << "\"requests\":"<< tmp_node.requests << ",";
            ss << "\"err_connect\":" << tmp_node.errors.connect <<",";
            ss << "\"err_read\":" << tmp_node.errors.read <<",";
            ss << "\"err_write\":" << tmp_node.errors.write<<",";
            ss << "\"err_status\":" << tmp_node.errors.status<<",";
            ss << "\"err_biz\":" << tmp_node.errors.biz<<",";
            ss << "\"err_timeout\":" << tmp_node.errors.timeout<< "},";

            // memset(&(iter->second), 0, sizeof(wrk_record_t));
        }

        string str = ss.str();
        if(str[str.length()-1]==',') {
            str.pop_back();
        }
        
        ss.str("");
        ss << str;
        ss << "}";
        return ss.str();
    }catch (exception& e) {
        cout<< e.what()<<endl;
    }
}

void WrkRecord::push_node(stringstream &ss, uint64_t x, stats_data *p) {
    // {
    //     "path":[{
    //         "x": 1564735298200,
    //         "data":{
    //             "connections:" 2,
    //             "qps": 4456.9930,
    //             "lancency": 100,
    //             "requests": 567,
    //             "error": {
    //                 "connect": 10,
    //                 "read":10,
    //                 "write":10,
    //                 "status":10,
    //                 "timeout":10
    //             }
    //         }
    //     },{

    //     }]
    // }

    ss << "{\"x\":"<<x << ",";
    ss << "\"connections\":"<<p->connections << ",";
    ss << "\"qps\":"<<p->qps << ",";
    ss << "\"latency\":"<<p->latency << ",";
    ss << "\"requests\":"<<p->requests << ",";
    ss << "\"err_connect\":" <<p->err.connect<<",";
    ss << "\"err_read\":" <<p->err.read<<",";
    ss << "\"err_write\":" <<p->err.write<<",";
    ss << "\"err_status\":" <<p->err.status<<",";
    ss << "\"err_biz\":" <<p->err.biz<<",";
    ss << "\"err_timeout\":" <<p->err.timeout<<"}";
}

void WrkRecord::push_series(stringstream &ss, series ser) {
    ss << "[";
    series::iterator x = ser.begin();
    for(int key=1; x!=ser.end(); x++, key++) {
        push_node(ss, x->first, x->second);
        ss << ",";
    }
    string str = ss.str();
    if(str[str.length()-1]==',') {
        str.pop_back();
    }
    
    ss.str("");
    ss << str << "]";
}


/**
 * @name: 
 * @test: 
 * @msg: 
 * @param {type} 
 * @return: 
 */
string WrkRecord::get_detail_json()
{
    stringstream ss;
    map<string, series>::iterator it = map_detail_data.begin();
    ss << "{";
    for(; it!=map_detail_data.end(); it++) {
        ss << "\""<< it->first.c_str() << "\":";
        push_series(ss, it->second);
        ss << ",";
    }

    string str = ss.str();
    if(str[str.length()-1]==',') {
        str.pop_back();
    }
    
    str += "}";
    return str;
}

string WrkRecord::get_stats_overview() 
{
    uint64_t complete  = record_total.complete;
    uint64_t rbytes    = record_total.rbytes;
    uint64_t wbytes    = record_total.wbytes;
    uint64_t requests  = record_total.requests;
    if(wbytes==0) { //没有收集到数据（还未发送任何请求）
        LOG_DEBUG("%s", "{}"); 
        return string("{}");
    }

    stringstream ss;

    uint64_t runtime_us = WrkCore::stop==1 ? WrkCore::cfg.stop_timestamp - WrkCore::cfg.start: WrkRecord::time_us() - WrkCore::cfg.start;
    long double runtime_s   = runtime_us / 1000000.0;
    long double req_per_s   = complete   / runtime_s;
    long double wbytes_per_s = wbytes    / runtime_s;
    long double rbytes_per_s = rbytes    / runtime_s;

    ss << "{\"Runtime\":" << runtime_s;
    ss << ",\"TotalRequests\":" << requests;
    ss << ",\"TotalCompletes\":" << complete;
    ss << ",\"TotalSendBytes\":" << wbytes;
    ss << ",\"TotalRecvBytes\":" << rbytes;
    ss << ",\"MeanQPS\":" << req_per_s;
    ss << ", \"SendRate\":" << wbytes_per_s;
    ss << ", \"RecvRate\":" << wbytes_per_s;

    long double mean  = stats_mean(statistics.latency);
    uint64_t max = statistics.latency->max;
    long double stdev = stats_stdev(statistics.latency, mean);
    ss << ",\"Latency\":{";
    ss << "\"Avg\":" << mean;
    ss << ",\"Stdev\":" << stdev;
    ss << ",\"Max\":" << max;
    // ss << ",\"+/-Stdev\":" << stats_within_stdev(statistics.latency, mean, stdev, 1);
    ss << "}";

    // mean  = stats_mean(statistics.requests);
    // max = statistics.requests->max;
    // stdev = stats_stdev(statistics.requests, mean);
    // ss << ",\"QPS\":{";
    // ss << "\"Avg\":" << mean;
    // ss << ",\"Stdev\":" << stdev;
    // ss << ",\"Max\":" << max;
    // // ss << ",\"+/-Stdev\":" << stats_within_stdev(statistics.requests, mean, stdev, 1);
    // ss << "}";

    ss << ",\"errors\":{";
    ss << "\"connect\":" << record_total.errors.connect;
    ss << ",\"read\":" << record_total.errors.read;
    ss << ",\"write\":" << record_total.errors.write;
    ss << ",\"timeout\":" << record_total.errors.timeout;
    ss << ",\"Non200\":" << record_total.errors.status;
    ss << ",\"biz\":" << record_total.errors.biz;
    ss << "}";

    ss << "}";

    LOG_DEBUG("\n%s",ss.str().c_str()); 

    return ss.str();

}

string WrkRecord::get_stats_percent() 
{
    stringstream ss;
    long double percentiles[] = { 25.0, 50.0, 75.0, 90.0, 99.0 };

    ss << "{";
    string str;
    map<string, statistics_t *>::iterator iter = map_sta.begin();
    for(; iter!=map_sta.end(); iter++) {
        stats_t *stats = iter->second->latency;
        ss << "\""<< iter->first.c_str() << "\":";
        ss << "{\"latency\":{";
        for (size_t i = 0; i < sizeof(percentiles) / sizeof(long double); i++) {
            long double p = percentiles[i];
            uint64_t n = stats_percentile(stats, p);
            string percent = std::to_string(p).substr(0,4).c_str();
            ss << "\"" << percent << "\":" << n << ",";
        }

        str = ss.str();
        if(str[str.length()-1]==',') {
            str.pop_back();
        }

        ss.clear();
        ss.str("");
        ss << str;
        ss << "}},";
    }

    str = ss.str();
    if(str[str.length()-1]==',') {
        str.pop_back();
    }

    ss.clear();
    ss.str("");
    ss << str;
    ss << "}";

    LOG_DEBUG(ss.str().c_str());
    return ss.str();
}

string WrkRecord::get_stats()
{
    stringstream ss;
    ss << "{";
    ss << "\"stop\":"<< WrkCore::stop;
    //    << get_detail_json();
    ss << ", \"detail\":" << get_moment_detail();
    ss << ",\"overview\":" << get_stats_overview();
    ss <<", \"percent\":" << get_stats_percent();
    ss <<"}\n";

    return ss.str();
}


//latency时,max=timeout(us)
stats_t *WrkRecord::stats_alloc(uint64_t max) {
    uint64_t limit = max + 1;
    stats_t *s = (stats_t *)zcalloc(sizeof(stats_t) + sizeof(uint64_t) * limit);
    s->limit = limit;
    s->min   = UINT64_MAX;
    return s;
}

void WrkRecord::stats_free(stats_t *stats) {
    zfree(stats);
}

void WrkRecord::stats_correct(stats_t *stats_ptr, int64_t expected) {
    for (uint64_t n = expected * 2; n <= stats_ptr->max; n++) {
        uint64_t count = stats_ptr->data[n];
        int64_t m = (int64_t) n - expected;
        int64_t i = 0;
        while (count>0 && m > expected) {
            stats_ptr->data[m] += count;
            stats_ptr->count += count;
            m -= expected;
        }
    }
}

long double WrkRecord::stats_mean(stats_t *stats) {
    if (stats->count == 0) return 0.0;

    uint64_t sum = 0;
    for (uint64_t i = stats->min; i <= stats->max; i++) {
        sum += stats->data[i] * i;
    }
    return sum / (long double) stats->count;
}

long double WrkRecord::stats_stdev(stats_t *stats, long double mean) {
    long double sum = 0.0;
    if (stats->count < 2) return 0.0;
    for (uint64_t i = stats->min; i <= stats->max; i++) {
        if (stats->data[i]) {
            sum += powl(i - mean, 2) * stats->data[i];
        }
    }
    return sqrtl(sum / (stats->count - 1));
}

long double WrkRecord::stats_within_stdev(stats_t *stats, long double mean, long double stdev, uint64_t n) {
    long double upper = mean + (stdev * n);
    long double lower = mean - (stdev * n);
    uint64_t sum = 0;

    for (uint64_t i = stats->min; i <= stats->max; i++) {
        if (i >= lower && i <= upper) {
            sum += stats->data[i];
        }
    }

    return (sum / (long double) stats->count) * 100;
}

uint64_t WrkRecord::stats_percentile(stats_t *stats, long double p) {
    uint64_t rank = round((p / 100.0) * stats->count + 0.5);
    uint64_t total = 0;
    for (uint64_t i = stats->min; i <= stats->max; i++) {
        total += stats->data[i];
        if (total >= rank) return i;
    }
    return 0;
}

uint64_t WrkRecord::stats_popcount(stats_t *stats) {
    uint64_t count = 0;
    for (uint64_t i = stats->min; i <= stats->max; i++) {
        if (stats->data[i]) count++;
    }
    return count;
}

uint64_t WrkRecord::stats_value_at(stats_t *stats, uint64_t index, uint64_t *count) {
    *count = 0;
    for (uint64_t i = stats->min; i <= stats->max; i++) {
        if (stats->data[i] && (*count)++ == index) {
            *count = stats->data[i];
            return i;
        }
    }
    return 0;
}


int WrkRecord::stats_record(stats_t *sta, uint64_t n)
{
    if (n >= sta->limit) return 0;
    __sync_fetch_and_add(&sta->data[n], 1); //return  stats->data[n]++;
    __sync_fetch_and_add(&sta->count, 1);
    uint64_t min = sta->min;
    uint64_t max = sta->max;
    while (n < min) {
        min = __sync_val_compare_and_swap(&sta->min, min, n);
    }

    while (n > max) {
        max = __sync_val_compare_and_swap(&sta->max, max, n);
    }

    return 1;
}

int WrkRecord::stats_record_latency(uint64_t n)
{
    return stats_record(statistics.latency, n);
}

